// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/expression/vector_function.h>

namespace kumo::pollux::functions {
    namespace {
        class NotFunction : public exec::VectorFunction {
        public:
            void apply(
                const SelectivityVector &rows,
                std::vector<VectorPtr> &args,
                const TypePtr & /* outputType */,
                exec::EvalCtx &context,
                VectorPtr &result) const override {
                POLLUX_CHECK_EQ(
                    args.size(), 1, "Incorrect number of arguments passed to not");
                const auto &input = args[0];

                POLLUX_CHECK_EQ(
                    input->type()->kind(), TypeKind::BOOLEAN, "Unsupported input type");

                BufferPtr negated;

                // Input may be constant or flat.
                if (input->is_constant_encoding()) {
                    bool value = input->as<ConstantVector<bool> >()->value_at(0);
                    negated =
                            AlignedBuffer::allocate<bool>(rows.end(), context.pool(), !value);
                } else {
                    negated = AlignedBuffer::allocate<bool>(rows.end(), context.pool());
                    auto *rawNegated = negated->asMutable<uint64_t>();

                    auto rawInput = input->as_flat_vector<bool>()->rawValues<uint64_t>();

                    memcpy(rawNegated, rawInput, bits::nbytes(rows.end()));
                    bits::negate(rawNegated, rows.end());
                }

                auto localResult = std::make_shared<FlatVector<bool> >(
                    context.pool(),
                    BOOLEAN(),
                    nullptr,
                    rows.end(),
                    negated,
                    std::vector<BufferPtr>{});

                context.moveOrCopyResult(localResult, rows, result);
            }

            exec::FunctionCanonicalName getCanonicalName() const override {
                return exec::FunctionCanonicalName::kNot;
            }

            static std::vector<std::shared_ptr<exec::FunctionSignature> > signatures() {
                // boolean -> boolean
                return {
                    exec::FunctionSignatureBuilder()
                    .returnType("boolean")
                    .argumentType("boolean")
                    .build()
                };
            }
        };
    } // namespace

    POLLUX_DECLARE_VECTOR_FUNCTION(
        udf_not,
        NotFunction::signatures(),
        std::make_unique<NotFunction>());
} // namespace kumo::pollux::functions
